package com.sh.data.engine.domain.datadev.offline.runner;

import cn.hutool.extra.spring.SpringUtil;
import com.alibaba.fastjson.JSON;
import com.google.common.collect.Lists;
import com.sh.data.engine.domain.base.model.datax.DataX;
import com.sh.data.engine.domain.base.model.datax.JobContentWriterParameter;
import com.sh.data.engine.domain.dataasset.service.TableProductService;
import com.sh.data.engine.domain.dataasset.service.TblPropService;
import com.sh.data.engine.domain.datadev.offline.model.enums.OfflineDevJobStatusEnum;
import com.sh.data.engine.domain.datadev.offline.service.DatadevJobService;
import com.sh.data.engine.domain.datadev.offline.service.DatadevOfflineSyncService;
import com.sh.data.engine.domain.datadev.offline.service.DatadevSyncFieldMappingService;
import com.sh.data.engine.domain.integration.datasource.model.domain.DataSourceDetailDomain;
import com.sh.data.engine.domain.integration.datasource.service.DataSourceService;
import com.sh.data.engine.domain.integration.offline.model.domain.OfflineSyncDomain;
import com.sh.data.engine.domain.integration.offline.service.DataXService;
import com.sh.data.engine.domain.util.VarParserUtil;
import com.sh.data.engine.repository.dao.dataasset.entity.TableProductEntity;
import com.sh.data.engine.repository.dao.dataasset.entity.TblPropEntity;
import com.sh.data.engine.repository.dao.datadev.entity.DatadevJobEntity;
import com.sh.data.engine.repository.dao.datadev.entity.DatadevMenuNodeEntity;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.io.FileUtils;
import org.apache.commons.lang3.StringUtils;

import java.io.*;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;

import static com.sh.data.engine.common.constants.Constants.LogFileExp;

@Slf4j
public class OfflineDevSyncTaskJobRunner extends OfflineDevBaseJobRunner {

    private static final String SKIP = "in ch.qos.logback";

    private static int PROCESS_FAILED_CODE = 143;
    private static int PROCESS_SUCCESS_CODE = 0;

    private volatile Process process = null;

    private DataXService dataXService = SpringUtil.getBean(DataXService.class);
    private TableProductService tableProductService = SpringUtil.getBean(TableProductService.class);
    private DatadevJobService offlineDevJobService = SpringUtil.getBean(DatadevJobService.class);
    private TblPropService tblPropService = SpringUtil.getBean(TblPropService.class);
    private DataSourceService dataSourceService = SpringUtil.getBean(DataSourceService.class);
    private DatadevOfflineSyncService datadevOfflineSyncService =
        SpringUtil.getBean(DatadevOfflineSyncService.class);
    private DatadevSyncFieldMappingService datadevSyncFieldMappingService =
        SpringUtil.getBean(DatadevSyncFieldMappingService.class);

    /***
     *
     * @param jobEntity
     * @param menuNodeEntity
     * @param parentNodeIdAndLineStatusMap
     * @param childNodeIdList
     * @param allJobNodeMap
     * @param executorService
     *
     * offline task
     */
    public OfflineDevSyncTaskJobRunner(
        DatadevJobEntity jobEntity,
        DatadevMenuNodeEntity menuNodeEntity,
        Map<Long, Integer> parentNodeIdAndLineStatusMap,
        List<Long> childNodeIdList,
        Map<Long, OfflineDevBaseJobRunner> allJobNodeMap,
        ExecutorService executorService) {
        super(
            jobEntity,
            menuNodeEntity,
            parentNodeIdAndLineStatusMap,
            childNodeIdList,
            allJobNodeMap,
            executorService);
    }

    @Override
    public void cancel() {
        process.destroy();
    }

    @Override
    public int runJob() throws Exception {

        String start = String.format("节点:[%s]开始执行", menuNodeEntity.getName());
        start = formatLogWithTime(start);
        this.writeLog(Lists.newArrayList(start));

        String finishedLog = String.format("节点:[%s]结束执行", menuNodeEntity.getName());
        OfflineSyncDomain offlineSyncDomain =
            datadevOfflineSyncService.getSyncTaskByNodeId(menuNodeEntity.getId());
        DataX dataX = dataXService.buildDataXJob(offlineSyncDomain);
        String jsonScript = setVars(dataX);
        final String dir = fileStorageConfiguration.getDir();
        String jsonPath =
            String.format(
                "%s/json/%s/%s.json", dir, LogFileExp.OFFLINE_DEV_JOB_LOG_EXP, jobEntity.getId());
        File file = new File(jsonPath);
        FileUtils.writeStringToFile(file, jsonScript, "UTF-8");

        String cmd = dataXConfig.getCmd(jsonPath);
        log.info("cmd:{}", cmd);

        process = Runtime.getRuntime().exec(cmd);

        CountDownLatch countDownLatch = new CountDownLatch(2);
        read(process.getInputStream(), countDownLatch);
        read(process.getErrorStream(), countDownLatch);

        int code = process.waitFor();
        log.info("exitCode:{}", code);
        countDownLatch.await();
        this.writeLog(Lists.newArrayList(finishedLog));

        // 如果运行成功且目标数据源是系统默认的
        DataSourceDetailDomain dataSourceDetailDomain =
            dataSourceService.getDataSourceDetailById(offlineSyncDomain.getTargetDsId(), false);
        if (code == PROCESS_SUCCESS_CODE) {
            // 产出信息记录
            if (null != dataSourceDetailDomain.getIsSystem()
                && dataSourceDetailDomain.getIsSystem().equals(1)
                && Objects.nonNull(jobEntity.getTaskFlowJobId())) {
                TblPropEntity tblPropByTblName =
                    tblPropService.getTblPropByTblName(
                        jobEntity.getProjectId(), offlineSyncDomain.getTargetTable());
                if (tblPropByTblName != null) {
                    TableProductEntity tableProductEntity = new TableProductEntity();
                    DatadevJobEntity jobServiceById =
                        offlineDevJobService.getById(jobEntity.getTaskFlowJobId());
                    offlineDevJobService.findById(jobEntity.getTaskFlowJobId());
                    tableProductEntity.setJobId(jobEntity.getId());
                    tableProductEntity.setTaskFlowId(jobServiceById.getNodeId());
                    tableProductEntity.setTblId(tblPropByTblName.getTblId());
                    tableProductEntity.setProjectId(jobEntity.getProjectId());
                    tableProductEntity.setSubTaskId(jobEntity.getNodeId());
                    tableProductService.insetTableProduct(Lists.newArrayList(tableProductEntity));
                }
            }
            // 任务执行完成后设置任务为已完成
            return OfflineDevJobStatusEnum.SUCCESS.getCode();
        } else if (code == PROCESS_FAILED_CODE) {
            return OfflineDevJobStatusEnum.CANCEL.getCode();
        }
        return OfflineDevJobStatusEnum.FAILURE.getCode();
    }

    private void read(InputStream in, CountDownLatch countDownLatch) {
        new Thread(
            () -> {
                try (BufferedReader reader = new BufferedReader(new InputStreamReader(in, "UTF-8"))) {
                    String line;
                    while ((line = reader.readLine()) != null) {
                        if (line.contains(SKIP)) {
                            continue;
                        }
                        List<String> logs = Lists.newArrayList(line);
                        writeLog(logs);
                    }
                } catch (IOException e) {
                    log.error(e.getMessage(), e);
                } finally {
                    countDownLatch.countDown();
                }
            })
            .start();
    }

    private String setVars(DataX dataX) {
        // 变量设置
        String where = dataX.getJob().getContent().get(0).getReader().getParameter().getWhere();
        JobContentWriterParameter parameter =
            dataX.getJob().getContent().get(0).getWriter().getParameter();
        if (StringUtils.isNotBlank(where)) {
            log.info("开始解析where条件, where:{}, parameter:{}", where, parameter);
            where =
                VarParserUtil.getWhere(
                    where,
                    dataX.getJob().getContent().get(0).getWriter().getName(),
                    parameter.getConnection().get(0).getJdbcUrl(),
                    parameter.getUsername(),
                    parameter.getPassword(),
                    parameter.getKrb5Conf(),
                    parameter.getKeyTabPath());
            dataX.getJob().getContent().get(0).getReader().getParameter().setWhere(where);
        }

        // 分区设置
        if (CollectionUtils.isNotEmpty(parameter.getPatitionValues())) {
            List<String> pts = Lists.newArrayList();
            for (String pt : parameter.getPatitionValues()) {
                pt = VarParserUtil.process(pt, null, null);
                pts.add(pt);
            }
            dataX.getJob().getContent().get(0).getWriter().getParameter().setPatitionValues(pts);
        }

        return JSON.toJSONString(dataX, true);
    }
}
